package rpc.server;

import rpc.intf.CommonSerializer;
import rpc.intf.RpcServer;
import rpc.intf.ServiceProvider;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.*;

@Slf4j
public class RpcSocketServer implements RpcServer {

    private int port;

    //线程池
    private final ExecutorService threadPool;
    //线程池参数
    private static final int CORE_POOL_SIZE = 5;
    private static final int MAXIMUM_POOL_SIZE = 50;
    private static final int KEEP_ALIVE_TIME = 60;
    private static final int BLOCKING_QUEUE_CAPACITY = 100;

    private RequestHandler requestHandler = new RequestHandler();

    //注册接口:方法
    private final ServiceProvider serviceRegistry;

    public RpcSocketServer(ServiceProvider serviceRegistry, int port) {
        this.port = port;
        this.serviceRegistry = serviceRegistry;
        BlockingQueue<Runnable> workingQueue = new ArrayBlockingQueue<>(BLOCKING_QUEUE_CAPACITY);
        ThreadFactory threadFactory = Executors.defaultThreadFactory();
        threadPool = new ThreadPoolExecutor(CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS, workingQueue, threadFactory);
    }

    //监听连接
    @Override
    public void start() {
        try(ServerSocket serverSocket = new ServerSocket(port)) {
            log.info("服务器正在启动......");
            Socket socket;
            while ((socket = serverSocket.accept()) != null) {
                log.info("消费者连接: {}:{}", socket.getInetAddress(), socket.getPort());

                threadPool.execute(new RequestHandlerThread(socket, requestHandler, serviceRegistry));
            }
        }catch (IOException e) {
            log.error("连接时有错误发生：", e);
        }
    }

    @Override
    public <T> void publishService(Object service, Class<T> serviceClass) {
        serviceRegistry.register(service);
        start();
    }


}
